package thread.callback.completablefuture;

import cn.hutool.core.thread.ThreadUtil;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;

import java.util.concurrent.*;
import java.util.function.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;

@Slf4j
public class CompletableFutureShuxueyunsuan {

    @Test
    public void test1() {

//    ExecutorService executorService = Executors.newFixedThreadPool(11);
//
//    // 有返回值
//    CompletableFuture<String> stringCompletableFuture = CompletableFuture.supplyAsync(
//            new Supplier<String>() {
//              @Override
//              public String get() {
//                try {
//                  Thread.sleep(3000);
//                } catch (InterruptedException e) {
//                  e.printStackTrace();
//                }
//                log.info( " done");
//                return "done";
//              }
//            }
//    );

//    //设置任务 成功完成后的回调
//    CompletableFuture<String> objectCompletableFuture = stringCompletableFuture.whenComplete(new BiConsumer<String, Throwable>() {
//      @Override
//      public void accept(String o, Throwable throwable) {
//        log.info(o + " world");
//      }
//    });
        log.info("start ");

        CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
            try {
                log.info("任务开始了 ");
                Thread.sleep(10000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            log.info("任务结束了 ");
            int i = 1 / 1;
            return i;
        });


//        try {
//            Thread.sleep(5000);
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }
        CompletableFuture<Integer> integerCompletableFuture = future.whenCompleteAsync(new BiConsumer<Integer, Throwable>() {
            @Override
            public void accept(Integer integer, Throwable throwable) {
                log.info("accept {}", integer);
            }
        });

        CompletableFuture<Integer> e1 = future.exceptionally(new Function<Throwable, Integer>() {
            @Override
            public Integer apply(Throwable throwable) {
                log.info("fail {}", throwable.getMessage());
                throwable.printStackTrace();
                return -1;
            }
        });


        log.info("end");

        for (int i = 0; i < 5; i++) {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

    }


    public void test2() {
        ExecutorService executorService = Executors.newFixedThreadPool(11);

//        //新建一个任务 无入参 无返回
//        CompletableFuture<Void> hello = CompletableFuture.runAsync(new Runnable() {
//            @Override
//            public void run() {
//                log.info("hello");
//            }
//        }, executorService);

        // 新建一个任务 有返回结果
        CompletableFuture<Object> hello = CompletableFuture.supplyAsync(new Supplier<Object>() {
            @Override
            public Object get() {
                ThreadUtil.sleep(1000);
                log.info("hello");
                return "hello";
            }
        }, executorService);

        log.info("main start");

        // 回调 有入参
        CompletableFuture<Void> world1 = hello.thenAcceptAsync(new Consumer<Object>() {
            @Override
            public void accept(Object o) {
                ThreadUtil.sleep(1000);
                log.info(String.valueOf(o) + " world1");
            }
        }, executorService);

        // 回调 有入参 有返回
        CompletableFuture<Object> world11 = hello.thenApplyAsync(new Function<Object, Object>() {
            @Override
            public Object apply(Object o) {
                ThreadUtil.sleep(1000);
                log.info(String.valueOf(o) + " world11");
                return String.valueOf(o) + " world11";
            }
        }, executorService);

        // 回调 无入参 无返回
        CompletableFuture<Void> world111 = hello.thenRunAsync(new Runnable() {
            @Override
            public void run() {
                ThreadUtil.sleep(1000);
                log.info("world111");
            }
        }, executorService);

        log.info("main end");

        // thenAccept  如果 CompletableFuture 任务没有完成，回调函数就在任务所在线程执行，  如果完成 回调函数就在当前线程执行
        hello.thenAccept(new Consumer<Object>() {
            @Override
            public void accept(Object o) {
                log.info(String.valueOf(o) + " world1111");
            }
        });


        log.info("get world1111 end ");

    }


    @Test
    public void thenCombineDemo() throws Exception {
        CompletableFuture<Integer> future1 =
                CompletableFuture.supplyAsync(new Supplier<Integer>() {
                    @Override
                    public Integer get() {
                        Integer firstStep = 10 + 10;
                        log.info("firstStep outcome is " + firstStep);
                        return firstStep;
                    }
                });
        CompletableFuture<Integer> future2 =
                CompletableFuture.supplyAsync(new Supplier<Integer>() {
                    @Override
                    public Integer get() {
                        Integer secondStep = 10 + 10;
                        log.info("secondStep outcome is " + secondStep);
                        return secondStep;
                    }
                });
        CompletableFuture<Integer> future3 = future1.thenCombine(future2,
                new BiFunction<Integer, Integer, Integer>() {
                    @Override
                    public Integer apply(
                            Integer step1OutCome, Integer step2OutCome) {
                        return step1OutCome * step2OutCome;
                    }
                });
        Integer result = future3.get();
        log.info(" outcome is " + result);
    }

    @Test
    public void test3() {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "Hello");
        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "CompletableFuture");
        CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> "Java");

        CompletableFuture<Void> combinedFuture = CompletableFuture.allOf(future1, future2, future3);

        Void join = combinedFuture.join();

        String result = Stream.of(future1, future2, future3)
                .map(CompletableFuture::join)
                .collect(Collectors.joining(" "));
        System.out.println(result);

    }

}
